import tensorflow as tf # 构建深度反欺诈模型 def build_fraud_detection_model(input_dim): model = tf.keras.Sequential([ tf.keras.layers.Dense(256, activation='relu', input_shape=(input_dim,)), tf.keras.layers.Dropout(0.3), tf.keras.layers.Dense(128, activation='relu'), tf.keras.layers.Dropout(0.3), tf.keras.layers.Dense(64, activation='relu'), tf.keras.layers.Dense(1, activation='sigmoid') # 欺诈概率 ]) model.compile( optimizer='adam', loss='binary_crossentropy', metrics=['AUC', 'precision', 'recall'] ) return model

import torch import torch.nn as nn from torch_geometric.nn import GCNConv class FraudGNN(nn.Module): def __init__(self, in_channels, hidden_channels, out_channels): super().__init__() self.conv1 = GCNConv(in_channels, hidden_channels) self.conv2 = GCNConv(hidden_channels, hidden_channels) self.conv3 = GCNConv(hidden_channels, out_channels) def forward(self, x, edge_index): # 第一层图卷积 x = self.conv1(x, edge_index) x = F.relu(x) x = F.dropout(x, p=0.5, training=self.training) # 第二层图卷积 x = self.conv2(x, edge_index) x = F.relu(x) x = F.dropout(x, p=0.5, training=self.training) # 输出层 x = self.conv3(x, edge_index) return F.log_softmax(x, dim=1)

import torch.nn as nn class BehaviorLSTM(nn.Module): def __init__(self, input_size, hidden_size, num_layers): super().__init__() self.lstm = nn.LSTM( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, batch_first=True, dropout=0.3 ) self.fc = nn.Linear(hidden_size, 1) self.sigmoid = nn.Sigmoid() def forward(self, x): # x shape: (batch, seq_len, input_size) lstm_out, (h_n, c_n) = self.lstm(x) # 取最后一个时间步的输出 last_output = lstm_out[:, -1, :] # 预测风险概率 risk_score = self.sigmoid(self.fc(last_output)) return risk_score

class AnomalyAutoencoder(nn.Module): def __init__(self, input_dim, encoding_dim): super().__init__() # 编码器 self.encoder = nn.Sequential( nn.Linear(input_dim, 128), nn.ReLU(), nn.Linear(128, 64), nn.ReLU(), nn.Linear(64, encoding_dim) ) # 解码器 self.decoder = nn.Sequential( nn.Linear(encoding_dim, 64), nn.ReLU(), nn.Linear(64, 128), nn.ReLU(), nn.Linear(128, input_dim) ) def forward(self, x): encoded = self.encoder(x) decoded = self.decoder(encoded) return decoded def detect_anomaly(self, x, threshold): reconstructed = self.forward(x) mse = torch.mean((x - reconstructed) ** 2, dim=1) return mse > threshold

import gym from stable_baselines3 import PPO # 定义额度管理环境 class CreditLimitEnv(gym.Env): def __init__(self): # 状态空间：客户特征 + 历史还款 + 当前额度使用率 self.observation_space = gym.spaces.Box(...) # 动作空间：提额/降额/保持，幅度0-50% self.action_space = gym.spaces.Discrete(11) def step(self, action): # 执行动作，更新额度 new_limit = self._adjust_limit(action) # 模拟客户行为（使用额度、还款） usage, repayment = self._simulate_customer() # 计算奖励：利息收入 - 逾期损失 - 运营成本 reward = self._calculate_reward(usage, repayment) return next_state, reward, done, info # 训练强化学习agent env = CreditLimitEnv() model = PPO("MlpPolicy", env, verbose=1) model.learn(total_timesteps=100000)

from openai import OpenAI client = OpenAI() def analyze_loan_application(application_text): prompt = f""" 你是一个专业的信贷风控专家。请分析以下贷款申请材料，识别潜在风险： 申请材料： {application_text} 请从以下维度分析： 1. 材料真实性（是否有伪造、PS痕迹） 2. 逻辑一致性（收入与职业是否匹配） 3. 还款能力评估 4. 潜在风险点 以JSON格式返回分析结果。 """ response = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0.3 ) return response.choices[0].message.content

from langchain.vectorstores import Chroma from langchain.embeddings import OpenAIEmbeddings from langchain.chains import RetrievalQA # 构建风控知识库 embeddings = OpenAIEmbeddings() vectordb = Chroma( persist_directory="./risk_knowledge", embedding_function=embeddings ) # RAG问答链 qa_chain = RetrievalQA.from_chain_type( llm=ChatOpenAI(model="gpt-4"), retriever=vectordb.as_retriever(search_kwargs={"k": 3}) ) # 查询 question = "针对月收入8000元、已有3笔小贷的客户，最高可授信多少？" answer = qa_chain.run(question)

# 自然语言输入 nl_rule = """ 拒绝同时满足以下条件的客户： 1. 年龄小于22岁 2. 近30天征信查询次数大于8次 3. 当前负债率超过60% """ # 大模型生成代码 generated_code = """ def check_risk(customer): if (customer['age'] < 22 and customer['credit_inquiry_30d'] > 8 and customer['debt_ratio'] > 0.6): return { 'decision': 'REJECT', 'reason': '年龄过小且多头借贷严重，负债率过高' } return {'decision': 'PASS'} """ # 自动测试 test_cases = load_historical_data() accuracy = evaluate_rule(generated_code, test_cases) print(f"规则准确率: {accuracy}")

def generate_rejection_explanation(customer_data, model_output): prompt = f""" 客户申请被拒绝，请生成一个友好、专业的解释说明。 客户信息：{customer_data} 模型输出：{model_output} 关键因素：{model_output['feature_importance']} 要求： 1. 使用通俗易懂的语言 2. 说明具体的拒绝原因 3. 提供改进建议 4. 语气友好，不要让客户感到被歧视 """ response = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}] ) return response.choices[0].message.content

def generate_collection_script(customer_profile): prompt = f""" 生成催收话术。客户信息如下： 姓名：{customer_profile['name']} 逾期天数：{customer_profile['overdue_days']}天 逾期金额：{customer_profile['overdue_amount']}元 客户类型：{customer_profile['customer_type']} # 能力型/意愿型/困难型 还款能力：{customer_profile['repayment_ability']} 历史沟通：{customer_profile['communication_history']} 要求： 1. 根据客户类型选择合适的语气和策略 2. 话术要礼貌专业，避免言语暴力 3. 提供具体的还款方案 4. 如果是困难型客户，体现同理心 5. 控制在200字以内 """ response = client.chat.completions.create( model="gpt-4", messages=[{"role": "user", "content": prompt}], temperature=0.7 ) return response.choices[0].message.content